package com.example.mq.mqserver.datacenter;

import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 用这个类来管理内存上数据的存储
 */
@Slf4j
public class MemoryDataManage {

    //考虑到线程安全问题，直接使用ConcurrentHashMap即可
    private DiskDataManage diskDataManage = new DiskDataManage();

    //缓存exchange:key-exchangeName,value-Exchange
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    //缓存queue:key-queueName,value-MSGQueue
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();

    //缓存binding:key-exchangeName,value-HashMap(key:queueName,value:Binding)
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    //缓存message:key-messageId(字符串类型),value-Message
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    //缓存队列与消息之间的关系:key-queueName,value-LinkedList(Message)
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    //缓存当前未被确认的消息（被消费者取走，但还没有应答的消息）：key-queueName,value-ConcurrentHashMap<messageId(字符串类型),Message>
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    private Map<String, ConsumerEnv> consumerEnvMap = new HashMap<>();


    public void insertExchange(Exchange exchange) {
        log.info("插入exchange成功");
        exchangeMap.put(exchange.getName(),exchange);
    }
    public void deleteExchange(String exchangeName) {
        log.info("删除exchange成功");
        exchangeMap.remove(exchangeName);
    }
    public Exchange getExchange(String exchangeName) {
        log.info("根据exchangeName获取exchange成功");
        return exchangeMap.get(exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(),queue);
        log.info("插入queue成功");
    }
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        log.info("删除queue成功");
    }
    public MSGQueue getQueue(String queueName) {
        log.info("根据queueName获取queue成功");
        return queueMap.get(queueName);
    }

    public void insertBinding(Binding binding) {
        //根据exchangeName查询bindingMap如果不存在就创建，否则不管
        //ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        //if(bindingMap == null) {
        //bindingMap = new ConcurrentHashMap<>();
        //bindingsMap.put(binding.getExchangeName(),bindingMap);
        //}
        //下面代码就是上述代码效果
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), t -> new ConcurrentHashMap<>());
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("当前队列已经被绑定,exchangeName:" + binding.getExchangeName() + " queueName:" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
            log.info("绑定成功,exchangeName:" + binding.getExchangeName() + " queueName:" + binding.getQueueName());
        }
    }

    public void deleteBinding(Binding binding) {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null) {
            throw new MqException("绑定不存在,exchangeName:" + binding.getExchangeName() + " queueName:" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("删除binding成功");
    }
    //这里写两个版本：
    //根据exchangeName和queueName获取唯一的Binding
    //根据exchangeName获取所有的Binding
    public Binding getBinding(String queueName,String exchangeName) {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    public void addMessage(Message message) {
        messageMap.put(message.getBasicProperties().getMessageId(),message);
        log.info("插入message成功");
    }
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        log.info("删除message成功");
    }
    public Message getMessage(String messageId) {
        log.info("根据messageId获取message成功");
        return messageMap.get(messageId);
    }

    //发送消息到指定队列
    public void sendMessage(String queueName, Message message) {
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queueName,t -> new LinkedList<>());
        synchronized (messages) {
            messages.add(message);
        }
        addMessage(message);
        log.info("[sendMessage] 发送消息到指定队列成功");
    }
    public void receiveMessage(String queueName, Message message) {
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queueName,t -> new LinkedList<>());
        synchronized (messages) {
            messages.add(message);
        }
        log.info("[receiveMessage] 发送消息到指定队列成功");
    }
    public Message pollMessage(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null || messages.size() == 0) {
            log.info("未找到对应队列");
            return null;
        }
        System.out.println("messageSize: " + messages.size());
        //采用头删的方式获取消息
        synchronized (messages) {
            log.info("获取队列中消息成功");
            return messages.pollFirst();
        }
    }

    //获取指定消息队列中的消息个数
    public int getCounts(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null || messages.size() == 0) {
            log.info("当前队列中消息个数为0");
            return 0;
        }
        synchronized (messages) {
            log.info("获取指定队列中消息个数成功");
            return messages.size();
        }
    }

    //添加未确认的消息（已经被取走了，但消费者还没有做出确认，确认确实已经进行了处理该消息）
    public void addMessageWaitAck(String queueName,Message message) {
        ConcurrentHashMap<String,Message> messages = queueMessageWaitAckMap.computeIfAbsent(queueName,t -> new ConcurrentHashMap<>());
        synchronized (messages) {
            messages.put(message.getBasicProperties().getMessageId(), message);
            log.info("添加未确认的消息成功");
        }
    }

    //删除未确认的消息（消费者这边已经确认了）
    public void removeMessageWaitAck(String queueName,String messageId) {
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if (messages == null && messages.size() == 0) {
            return;
        }
        synchronized (messages) {
            messages.remove(messageId);
            log.info("删除未确认的消息成功");
        }
    }

    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,Message message) {
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if (messages == null && messages.size() == 0) {
            log.info("获取指定队列未确认消息失败");
            return null;
        }
        log.info("获取指定队列未确认消息成功");
        return messages.get(message.getBasicProperties().getMessageId());
    }

    //加载data文件中的数据到内存中（重启后应该做的工作）
    public void recovery(DiskDataManage diskDataManage) throws IOException, ClassNotFoundException {
        //1.删除内存中的所有数据
        messageMap.clear();
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        queueMessageMap.clear();
        //2.恢复queueMap中数据
        List<MSGQueue> queues = diskDataManage.selectQueues();
        for(MSGQueue queue : queues) {
            queueMap.put(queue.getName(),queue);
        }
        log.info("queueMap中数据恢复完成");
        //3.恢复messageMap和queueMessageMap中的数据
        for(MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataManage.loadMessages(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for(Message message : messages) {
                messageMap.put(message.getBasicProperties().getMessageId(),message);
            }
        }
        //4.恢复exchangeMap中的数据
        List<Exchange> exchanges = diskDataManage.selectExchanges();
        for(Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(),exchange);
        }
        //5.恢复bindsingMap中的数据
        List<Binding> bindings = diskDataManage.selectBindings();
        for(Binding binding : bindings) {
            ConcurrentHashMap<String,Binding> bind = new ConcurrentHashMap<>();
            bind.put(binding.getQueueName(),binding);
            bindingsMap.put(binding.getExchangeName(),bind);
        }
        log.info("bindsingMap中数据恢复完成");

        //注意："未被确认的消息"不需要从硬盘上恢复
        // 在等待ack的过程中（服务器），一旦当服务器重启了过后这部分数据就会变成"未取走的消息"
        //未取走的消息就是在硬盘上存储的，此时消费者需要重新从对列中取
    }

    public void addConsumer(String consumerTag,ConsumerEnv consumerEnv) {
        consumerEnvMap.put(consumerTag,consumerEnv);
    }

    public ConsumerEnv getConsumer(String consumerTag) {
        return consumerEnvMap.get(consumerTag);
    }
}
